package com.idroc.coelacanth.service.servant.bdb;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.idroc.coelacanth.core.storage.Cursor;
import com.idroc.coelacanth.core.storage.Database;
import com.idroc.coelacanth.core.storage.Record;
import com.idroc.coelacanth.core.storage.StorageKey;
import com.idroc.coelacanth.core.storage.StorageValue;
import com.idroc.coelacanth.core.storage.support.KeyValueWrapperRecord;
import com.idroc.coelacanth.core.storage.support.KeyValues;
import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.Environment;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.OperationStatus;

public class BerkeleyDatabase implements Database {
	
	private static final Logger log = LoggerFactory.getLogger(BerkeleyDatabase.class);
	 
	private Environment env = null;
	private String dbName = null;
	private com.sleepycat.je.Database bdb = null;
	
	private int uncommitedCount = 0;
	
	BerkeleyDatabase(String dbName, Environment env) {

		this.dbName = dbName;
		this.env = env;
	}

	public void close() {
		
		try {
			bdb.close();
		} catch (DatabaseException ex) {  
			throw new RuntimeException(ex);
		}
	}

	public long count() { 
		 
		try {
			return bdb.count();
		} catch (DatabaseException ex) {
			throw new RuntimeException(ex);
		} 
	}

	public boolean delete(StorageKey key) {

		DatabaseEntry k = new DatabaseEntry(key.getBytes());
		OperationStatus os = null;
		
		try {
			os = bdb.delete(null, k); 
		} catch (DatabaseException dbe) {
			log.error("节点删除数据异常, key[{}]", key); 
		}
		return  (os == OperationStatus.SUCCESS) ;
	}

	public Record get(StorageKey key) {
		
		DatabaseEntry k = new DatabaseEntry(key.getBytes());
		DatabaseEntry v = new DatabaseEntry();
		OperationStatus os = null;
		try {
			os = bdb.get(null, k, v, LockMode.DEFAULT);
		} catch (DatabaseException dbe) {
			log.error("数据库查找记录异常, key[{}]", key); 
		}
		if (os == OperationStatus.SUCCESS) {
			return new KeyValueWrapperRecord(key, KeyValues.newByteArrayValue(v.getData()));
		} else {
			return null;
		}
	}

	public Cursor getCursor() {
		
		return new BerkeleyCursor(bdb);
	}

	public void open() {
		try {			  
			DatabaseConfig dbConfig = new DatabaseConfig();
		    dbConfig.setAllowCreate(true); 
		    dbConfig.setDeferredWrite(true);
		    bdb = env.openDatabase(null, dbName, dbConfig);  
		} catch (DatabaseException ex) {
			log.error("打开节点异常", ex); 
			throw new RuntimeException(ex);
		}
	}

	public boolean save(StorageKey key, StorageValue value) {

		DatabaseEntry k = new DatabaseEntry(key.getBytes());
		DatabaseEntry v = new DatabaseEntry(value.getBytes());
		try {
			OperationStatus os = bdb.put(null, k, v);
			if (os == OperationStatus.SUCCESS) {
				// 每1000条记录同步数据库到磁盘.
				if (uncommitedCount >= 1000) {
					bdb.sync();
					uncommitedCount = 0;
				}
				return true;
			} 
		} catch (DatabaseException ex) {
			log.error("节点保存数据异常, key[{}]", key); 
			throw new RuntimeException(ex);
		}
		return false;
	}

}
